﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Adrian.Algorithm;
using Ddbs.QE.Message;
using System.Data;

namespace Ddbs.QE.Parser.Front
{

    public enum AlgebraTreeNodeType
    {
        PROJECTION = 1,  //COLUMNS

        JOIN,       //COLUMNS
        SEMIJOIN,   //COLUMNS
        PRODUCT,

        UNION,
        //  DIFF,
        INTERSECTION,

        SELECTION,  //CONDITION

        TABLE,   // Table, mostly for primary algebra tree
        HFRAGMENT,    //Fragment, with fragmentation infomation 
        VFRAGMENT    //Fragment, with fragmentation infomation

    }
    public class AlgebraTreeNode
    {
        public AlgebraTreeNode()
        {
            children = new List<AlgebraTreeNode>();
        }

        private AlgebraTreeNode parent;

        public AlgebraTreeNode Parent
        {
            get { return parent; }
            set { parent = value; }
        }
        private IList<AlgebraTreeNode> children;

        public IList<AlgebraTreeNode> Children
        {
            get { return children; }
            set { children = value; }
        }
        private AlgebraTreeNodeType type;

        public AlgebraTreeNodeType Type
        {
            get { return type; }
            set { type = value; }
        }

        virtual public AlgebraTreeNode dup()
        {
            AlgebraTreeNode copy = new AlgebraTreeNode();
            copy.parent = parent;
            copy.type = type;
            foreach (AlgebraTreeNode childNode in children)
            {
                AlgebraTreeNode copyChild = childNode.dup();
                copyChild.parent = copy;
                copy.children.Add(copyChild);
            }
            return copy;
        }
        private Serialization.Site execSite;

        public Serialization.Site ExecSite
        {
            get { return execSite; }
            set { execSite = value; }
        }
        private int totalSize;

        public int TotalSize
        {
            get { return totalSize; }
            set { totalSize = value; }
        }
        private int messageTrans;

        public int MessageTrans
        {
            get { return messageTrans; }
            set { messageTrans = value; }
        }

        private int cost;

        public int Cost
        {
            get { return cost; }
            set { cost = value; }
        }

    }
    public class ProjectionNode : AlgebraTreeNode
    {
        IList<Ddbs.QE.Parser.Front.Serialization.Column> columns;

        public IList<Ddbs.QE.Parser.Front.Serialization.Column> Columns
        {
            get { return columns; }
            set { columns = value; }
        }
        public ProjectionNode()
        {
            Type = AlgebraTreeNodeType.PROJECTION;
            columns = new List<Serialization.Column>();
        }
        public ProjectionNode(IList<Ddbs.QE.Parser.Front.Serialization.Column> columns)
        {
            Type = AlgebraTreeNodeType.PROJECTION;
            this.columns = columns;
        }


        override public AlgebraTreeNode dup()
        {

            ProjectionNode copy = new ProjectionNode();
            copy.Parent = Parent;
            copy.Type = AlgebraTreeNodeType.PROJECTION;
            foreach (AlgebraTreeNode childNode in Children)
            {
                AlgebraTreeNode copyChild = childNode.dup();
                copyChild.Parent = copy;
                copy.Children.Add(copyChild);
            }
            foreach (Ddbs.QE.Parser.Front.Serialization.Column column in columns)
            {
                copy.columns.Add(column.dup());
            }
            return copy;
        }


    }
    public class SelectionNode : AlgebraTreeNode
    {
        Ddbs.QE.Parser.Front.Serialization.Condition cond;

        public Ddbs.QE.Parser.Front.Serialization.Condition Cond
        {
            get { return cond; }
            set { cond = value; }
        }
        public SelectionNode()
        {
            Type = AlgebraTreeNodeType.SELECTION;
        }
        public SelectionNode(Ddbs.QE.Parser.Front.Serialization.Condition cond)
        {
            Type = AlgebraTreeNodeType.SELECTION;
            this.cond = cond;
        }

        override public AlgebraTreeNode dup()
        {

            SelectionNode copy = new SelectionNode();
            copy.Parent = Parent;
            copy.Type = AlgebraTreeNodeType.SELECTION;
            foreach (AlgebraTreeNode childNode in Children)
            {
                AlgebraTreeNode copyChild = childNode.dup();
                copyChild.Parent = copy;
                copy.Children.Add(copyChild);
            }
            copy.cond = cond;
            return copy;
        }

    }
    public class JoinNode : AlgebraTreeNode
    {

        IList<Ddbs.QE.Parser.Front.Serialization.Column> columns;

        public IList<Ddbs.QE.Parser.Front.Serialization.Column> Columns
        {
            get { return columns; }
            set { columns = value; }
        }
        public JoinNode()
        {
            Type = AlgebraTreeNodeType.JOIN;
            columns = new List<Serialization.Column>();
        }
        public JoinNode(IList<Ddbs.QE.Parser.Front.Serialization.Column> columns)
        {
            Type = AlgebraTreeNodeType.JOIN;
            this.columns = columns;
        }

        override public AlgebraTreeNode dup()
        {

            JoinNode copy = new JoinNode();
            copy.Parent = Parent;
            copy.Type = AlgebraTreeNodeType.JOIN;
            foreach (AlgebraTreeNode childNode in Children)
            {
                AlgebraTreeNode copyChild = childNode.dup();
                copyChild.Parent = copy;
                copy.Children.Add(copyChild);
            }
            foreach (Ddbs.QE.Parser.Front.Serialization.Column column in columns)
            {
                copy.columns.Add(column.dup());
            }
            return copy;
        }

    }
    public class SemiJoinNode : AlgebraTreeNode
    {
        public SemiJoinNode()
        {
            Type = AlgebraTreeNodeType.SEMIJOIN;
        }
        public SemiJoinNode(IList<Ddbs.QE.Parser.Front.Serialization.Column> columns)
        {
            Type = AlgebraTreeNodeType.SEMIJOIN;
            this.columns = columns;
        }
        IList<Ddbs.QE.Parser.Front.Serialization.Column> columns;

        public IList<Ddbs.QE.Parser.Front.Serialization.Column> Columns
        {
            get { return columns; }
            set { columns = value; }
        }


        override public AlgebraTreeNode dup()
        {

            SemiJoinNode copy = new SemiJoinNode();
            copy.Parent = Parent;
            copy.Type = AlgebraTreeNodeType.SEMIJOIN;
            foreach (AlgebraTreeNode childNode in Children)
            {
                AlgebraTreeNode copyChild = childNode.dup();
                copyChild.Parent = copy;
                copy.Children.Add(copyChild);
            }
            foreach (Ddbs.QE.Parser.Front.Serialization.Column column in columns)
            {
                copy.columns.Add(column.dup());
            }
            return copy;
        }

    }
    public class TableNode : AlgebraTreeNode
    {

        public TableNode(Ddbs.QE.Parser.Front.Serialization.Table tbl)
        {
            Type = AlgebraTreeNodeType.TABLE;
            table = tbl;
        }

        Ddbs.QE.Parser.Front.Serialization.Table table;

        public Ddbs.QE.Parser.Front.Serialization.Table Table
        {
            get { return table; }
            set { table = value; }
        }

        override public AlgebraTreeNode dup()
        {

            TableNode copy = new TableNode(table);
            copy.Parent = Parent;
            copy.Type = AlgebraTreeNodeType.TABLE;
            foreach (AlgebraTreeNode childNode in Children)
            {
                AlgebraTreeNode copyChild = childNode.dup();
                copyChild.Parent = copy;
                copy.Children.Add(copyChild);
            }

            return copy;
        }



    }
    public class FragmentNode : AlgebraTreeNode
    {
        
        public FragmentNode(Ddbs.QE.Parser.Front.Serialization.Fragment frag, AlgebraTreeNodeType type)
        {
            Type = AlgebraTreeNodeType.TABLE;
            this.frag = frag;
            this.Type = type;
        }
        Ddbs.QE.Parser.Front.Serialization.Fragment frag;

        public Ddbs.QE.Parser.Front.Serialization.Fragment Frag
        {
            get { return frag; }
            set { frag = value; }
        }

        override public AlgebraTreeNode dup()
        {

            FragmentNode copy = new FragmentNode(frag, Type);
            copy.Parent = Parent;
            copy.Type = Type;
            copy.frag = frag;
            foreach (AlgebraTreeNode childNode in Children)
            {
                AlgebraTreeNode copyChild = childNode.dup();
                copyChild.Parent = copy;
                copy.Children.Add(copyChild);
            }

            return copy;
        }

    }
    public class AlgebraTree
    {


        


        private AlgebraTreeNode root; //projection
        private SelectionNode select; //select
        private AlgebraTreeNode join; //join or table



        public AlgebraTreeNode Root
        {
            get { return root; }
            //set { root = value; }
        }
        public AlgebraTree(Magic.Framework.OQL.SQLContext context)
        {
            root = new ProjectionNode(context.Columns);
            SelectionNode selection = null;

            Dictionary<string, List<Serialization.Condition>> tableName2Selection = new Dictionary<string, List<Serialization.Condition>>();
            List<Serialization.Column> joincolumns = new List<Serialization.Column>();
            if (context.Cond != null)
            {
                pushDownNotHelper(context.Cond, false);
                Serialization.Condition.normalize(context.Cond);
                int result = Serialization.Condition.tryEvaluate(context.Cond);
                if (result == 0)
                {
                    //normal
                    selection = new SelectionNode(context.Cond);
                    selection.Parent = root;
                    root.Children.Add(selection);

                }else if(result == -1)
                {
                    //always false
                    throw new Exception("no rows");

                }
                // 1 every tuple satisfys the condition
                select = selection;
                collectConds(select.Cond, tableName2Selection);
                collectJoinConds(select.Cond, joincolumns);
            }



            //bool first = true, second = true;
            AlgebraTreeNode cur = null;
            //AlgebraTreeNode tmpProduct = null;

            List<AlgebraTreeNode> phyPlanForFrag = new List<AlgebraTreeNode>();
            foreach (Ddbs.QE.Parser.Front.Serialization.Table table in context.Tables)
            {
                phyPlanForFrag.Add(createSeletedFrags(table, tableName2Selection));
            }

            cur = createdJoinedTables(phyPlanForFrag, joincolumns);
            /*
            //TableNode firstTable = null;
            foreach (Ddbs.QE.Parser.Front.Serialization.Table table in context.Tables)
            {
                if (first)
                {

                    tmpProduct = new AlgebraTreeNode();
                    tmpProduct.Type = AlgebraTreeNodeType.JOIN;

                    AlgebraTreeNode tmpTable = createSeletedFrags(table, tableName2Selection);
                    
                    tmpProduct.Children.Add(tmpTable);
                    tmpTable.Parent = tmpProduct;
                    cur = tmpTable;

                    //firstTable = new TableNode(table);
                    //cur = firstTable;
                    first = false;
                }
                else
                {
                    //first = false;
                    if (second)
                    {
                        cur = tmpProduct;
                        second = false;
                    }
                    AlgebraTreeNode tmpTable =  createSeletedFrags(table, tableName2Selection);
                    cur.Children.Add(tmpTable);
                    tmpTable.Parent = cur;

                }
            }*/
            AlgebraTreeNode curRoot = null;
            if (selection == null)
            {
                curRoot = root;
            }
            else
            {
                curRoot = selection;
            }
            curRoot.Children.Add(cur);
            cur.Parent = curRoot;
            determineExecSite(root);
        }

       
        private void pushDownNotHelper(Serialization.Condition condition, bool meetNot)
        {


            if (meetNot)
            {
                switch (condition.Type)
                {
                    case Serialization.ConditionType.AND:
                        condition.Type = Serialization.ConditionType.OR;
                        foreach (Serialization.Condition child in condition.Children)
                        {
                            pushDownNotHelper(child, true);
                        }
                        break;
                    case Serialization.ConditionType.OR:
                        condition.Type = Serialization.ConditionType.AND;
                        foreach (Serialization.Condition child in condition.Children)
                        {
                            pushDownNotHelper(child, true);
                        }
                        break;
                    case Serialization.ConditionType.NOT:
                        //condition
                        //tmpRoot = pushDownNotHelper(condition.Children.First(), false);
                        pushDownNotHelper(condition.Children.First(), false);
                        condition.Type = condition.Children.First().Type;
                        condition.Children = condition.Children.First().Children;

                        break;
                    case Serialization.ConditionType.EQ:
                        condition.Type = Serialization.ConditionType.NEQ;

                        break;
                    case Serialization.ConditionType.GE:
                        condition.Type = Serialization.ConditionType.LT;

                        break;

                    case Serialization.ConditionType.GT:
                        condition.Type = Serialization.ConditionType.LE;

                        break;
                    case Serialization.ConditionType.LE:
                        condition.Type = Serialization.ConditionType.GT;

                        break;
                    case Serialization.ConditionType.LT:
                        condition.Type = Serialization.ConditionType.GE;

                        break;
                    case Serialization.ConditionType.NEQ:
                        condition.Type = Serialization.ConditionType.EQ;

                        break;
                    default:
                        break;
                }
            }
            else
            {
                switch (condition.Type)
                {
                    case Serialization.ConditionType.AND:
                        //tmpRoot.Type = Serialization.ConditionType.AND;
                        foreach (Serialization.Condition child in condition.Children)
                        {
                            pushDownNotHelper(child, false);
                        }
                        break;
                    case Serialization.ConditionType.OR:
                        //tmpRoot.Type = Serialization.ConditionType.OR;
                        foreach (Serialization.Condition child in condition.Children)
                        {
                            pushDownNotHelper(child, false);
                        }
                        break;
                    case Serialization.ConditionType.NOT:

                        pushDownNotHelper(condition.Children.First(), true);
                        condition.Type = condition.Children.First().Type;
                        condition.Children = condition.Children.First().Children;

                        break;

                    default:
                        break;
                }

            }


        }



        private void splay()
        {
            if (root.Children != null && root.Children.Count != 0)
            {
                if (root.Children.First().Type == AlgebraTreeNodeType.SELECTION)
                {
                    SelectionNode selection = root.Children.First() as SelectionNode;

                    AlgebraTreeNode newNode = splaySelection(selection);
                    root.Children.RemoveAt(0);
                    root.Children.Add(newNode);
                    newNode.Parent = root;
                }
            }
        }


        private AlgebraTreeNode splaySelection(SelectionNode selection)
        {
            Serialization.Condition condition = selection.Cond;
            AlgebraTreeNode newNode = splayHelper(condition, selection);
            return newNode;
        }

        private AlgebraTreeNode splayHelper(Serialization.Condition cond, AlgebraTreeNode original)
        {
            AlgebraTreeNode tmp = null;
            AlgebraTreeNode cur = null;
            SelectionNode tmpSelection = null;
            switch (cond.Type)
            {
                case Serialization.ConditionType.AND:
                    //tmp = new AlgebraTreeNode();
                    //tmp.Type = AlgebraTreeNodeType.INTERSECTION;
                    bool firstAnd = true;


                    foreach (Serialization.Condition childCond in cond.Children)
                    {
                        tmpSelection = new SelectionNode();
                        tmpSelection.Cond = childCond;
                        if (firstAnd)
                        {
                            firstAnd = false;
                            tmp = tmpSelection;
                            cur = tmpSelection;
                        }
                        else
                        {
                            cur.Children.Add(tmpSelection);
                            tmpSelection.Parent = cur;
                            cur = tmpSelection;
                        }
                    }
                    foreach (AlgebraTreeNode childNode in original.Children)
                    {
                        cur.Children.Add(childNode.dup());
                    }
                    tmp = splaySelection(tmp as SelectionNode);

                    break;
                case Serialization.ConditionType.OR:

                    if (Serialization.Condition.isSimpleOr(cond))
                    {
                        return new SelectionNode(cond);
                    }


                    tmp = new AlgebraTreeNode();
                    tmp.Type = AlgebraTreeNodeType.UNION;
                    foreach (Serialization.Condition childCond in cond.Children)
                    {
                        tmpSelection = new SelectionNode();
                        tmpSelection.Cond = childCond;


                        foreach (AlgebraTreeNode childNode in original.Children)
                        {
                            tmpSelection.Children.Add(childNode.dup());
                        }
                        cur = splayHelper(childCond, tmpSelection);
                        cur.Parent = tmp;
                        tmp.Children.Add(cur);
                    }

                    break;
                /* case Serialization.ConditionType.EQ:
                 break;
                 case Serialization.ConditionType.NEQ:
                 break;
                 case Serialization.ConditionType.LE:
                 break;
                 case Serialization.ConditionType.LT:
                 break;
                 case Serialization.ConditionType.GT:
                 break;
                 case Serialization.ConditionType.GE:
                 break;
                 case Serialization.ConditionType.INTVAL:
                 break;
                 case Serialization.ConditionType.STRVAL:
                 break;*/
                default:
                    tmp = original.dup();
                    break;
            }


            return tmp;
        }

        private void walk(SelectionNode selection, ProjectionNode projection)
        {
            List<Serialization.Condition> conditions = new List<Serialization.Condition>();
            walk_helper(selection, conditions, projection);

        }
        private void walk_helper(AlgebraTreeNode root, List<Serialization.Condition> conditions, ProjectionNode projection)
        {
            
            switch(root.Type)
            {
                case AlgebraTreeNodeType.SELECTION:
                    conditions.Add((root as SelectionNode).Cond);
                    if (root.Children != null && root.Children.Count != 0)
                    {
                        walk_helper(root.Children.First(), conditions, projection);
                    }
                    break;
                
                case AlgebraTreeNodeType.UNION:
                    foreach (AlgebraTreeNode child in root.Children)
                    {
                        walk_helper(child, conditions, projection);
                    }
                    break;
                case AlgebraTreeNodeType.PROJECTION:
                    //to do
                    break;
                case AlgebraTreeNodeType.JOIN:
                    //
                    List<Serialization.Condition> joinConds = new List<Serialization.Condition>();
                    List<Serialization.Column> selectColumns = new List<Serialization.Column>();
                        

                    


                    foreach (Serialization.Condition cond in conditions)
                    {
                        if (Serialization.Condition.isJoinCond(cond))
                        {
                            joinConds.Add(cond);
                        }
                        extractColumns(cond, selectColumns);
                    }
                    
                    foreach (AlgebraTreeNode child in root.Children)
                    {
                        List<Serialization.Condition> tmpConds = new List<Serialization.Condition>();
                        List<Serialization.Column> tmpColumns = new List<Serialization.Column>();
                        foreach (Serialization.Column c in projection.Columns)
                        {
                            if (applicable(c, child))
                            {
                                tmpColumns.Add(c);
                            }
                        }
                        foreach (Serialization.Column c in selectColumns)
                        {
                            if (applicable(c, child) && (!tmpColumns.Contains(c)) )
                            {
                                tmpColumns.Add(c);
                            }
                        }


                        foreach (Serialization.Condition cond in conditions)
                        {
                            if (!Serialization.Condition.isJoinCond(cond))
                            {
                                if (applicable(cond, child  as TableNode))
                                {
                                    tmpConds.Add(cond);
                                }
                            }
                        }


                        ProjectionNode tmpProjection = new ProjectionNode(tmpColumns);

                        AlgebraTreeNode preRoot = tmpProjection, firstSelection = null;
                        
                        bool isFirstSelection = true;

                        foreach (Serialization.Condition cond in tmpConds)
                        {
                            SelectionNode selection  = new SelectionNode(cond);
                            selection.Parent = preRoot;
                            if (isFirstSelection)
                            {
                                firstSelection = selection;
                                isFirstSelection = false;
                            }
                            else {
                                preRoot.Children.Add(selection);
                            }
                            preRoot = selection;

                        }
                        preRoot.Children.Add(child);

                    }
                    break;
                case AlgebraTreeNodeType.TABLE:
                    break;
                default:
                    break;
            }


        }

        private static bool applicable(Serialization.Condition cond, TableNode table)
        {
            if (Serialization.Condition.isPred(cond))
            {
                if (cond.Children.First().Type == Serialization.ConditionType.COLUMN)
                {
 
                }
            }
            else if(Serialization.Condition.isSimpleOr(cond)){
                
            }
            return false;
        }

        private static bool applicable(Serialization.Column column, AlgebraTreeNode node)
        {
            return false;
        }

        private static void extractColumns(Serialization.Condition cond, List<Serialization.Column> columns)
        {
            if (Serialization.Condition.isPred(cond))
            {

                foreach (Serialization.Condition child in cond.Children)
                {
                    if (child.Type == Serialization.ConditionType.COLUMN)
                    {
                        Serialization.ColumnVal columnval = child as Serialization.ColumnVal;
                        columns.Add(columnval.ColumnRef);
                        
                    }
                }

                
            }
            else if(Serialization.Condition.isSimpleOr(cond)){
                foreach (Serialization.Condition child in cond.Children)
                {
                    extractColumns(cond, columns);
                }
            }
        }


        private static AlgebraTreeNode createdJoinedTables(List<AlgebraTreeNode> tables, List<Serialization.Column> columns)
        {
            AlgebraTreeNode tmp=null;
            AlgebraTreeNode tmpJoin = new JoinNode();

            Permutations<AlgebraTreeNode> perms = new Permutations<AlgebraTreeNode>(tables);
            int factor = tables.Count ;
            IEnumerator<IList<AlgebraTreeNode>> enumerator = perms.GetEnumerator();
            AlgebraTreeNode curBestPhyPlan = null;
            double curBestCost =Double.MaxValue;
            while (enumerator.MoveNext())
            {
                //logical plan
                IList<AlgebraTreeNode> plan = enumerator.Current;
                int l = 0, r = factor - 1;
                AlgebraTreeNode phyPlan = createBestPhysicalPlan(plan, l, r, columns);
                //catalan(factor+1) physical plan for each logical plan
                if (phyPlan.Cost < curBestCost)
                {
                    curBestCost = phyPlan.Cost;
                    curBestPhyPlan = phyPlan;
                }
                
             }

            return curBestPhyPlan;
        }

        private static AlgebraTreeNode createSeletedFrags(Serialization.Table table, Dictionary<string, List<Serialization.Condition>> tableName2Selection)
        {
            AlgebraTreeNode tmp = null;
            SelectionNode selection = null;
            if (tableName2Selection.ContainsKey(table.Name))
            {
                selection = new SelectionNode();
                List<Serialization.Condition> subConds = tableName2Selection[table.Name];
                if (subConds.Count > 1)
                {
                    selection.Cond = new Serialization.Condition();
                    selection.Cond.Type = Serialization.ConditionType.AND;
                    selection.Cond.Children.AddRange(subConds);
                }
                selection.Cond = subConds.First();
                tmp = selection;
            }
            else { 
                
            
            }
            if (table.Fragments.First() is Serialization.VFragment)
            {
                JoinNode tmpJoin = new JoinNode();

                foreach (Serialization.Fragment f in table.Fragments)
                {
                    Serialization.VFragment vf = f as Serialization.VFragment;
                    //to do if() for projection
                    FragmentNode tmpVF = new FragmentNode(vf, AlgebraTreeNodeType.VFRAGMENT);
                    tmpVF.ExecSite = vf.Sites.First();
                    tmpVF.MessageTrans = 0;
                    tmpVF.TotalSize = (vf.RowSize*vf.RowCount);
                    tmpVF.Cost = tmpVF.TotalSize+ tmpVF.MessageTrans;
                    tmpJoin.Columns.Add(vf.JoinColumn);
                    if(false)
                    //if (selection != null)
                    {

                        SelectionNode selectionDup = selection.dup() as SelectionNode;
                        tmpVF.Parent = selectionDup;
                        selectionDup.Children.Add(tmpVF);
                        selectionDup.Parent = tmpJoin;
                        selection.ExecSite = tmpVF.ExecSite;
                        selection.MessageTrans=tmpVF.MessageTrans;
                        selection.TotalSize = tmpVF.TotalSize;
                        selection.Cost = tmpVF.Cost;


                        tmpJoin.Children.Add(selectionDup);
                        

                    }
                    else
                    {
                        tmpVF.Parent = tmpJoin;
                        tmpJoin.Children.Add(tmpVF);
                        
                    }
                }
                if (tmpJoin.Children.Count > 1)
                {
                    double curBestCost=Double.MaxValue;
                    AlgebraTreeNode curBest =null;
                    foreach(AlgebraTreeNode child in tmpJoin.Children)
                    {
                        if(child.Cost<curBestCost)
                        {
                            curBest = child;
                            curBestCost = child.Cost;
                        }
                        tmpJoin.TotalSize += child.TotalSize;
                    }
                    tmpJoin.ExecSite = curBest.ExecSite;
                    tmpJoin.MessageTrans = tmpJoin.TotalSize - curBest.TotalSize;
                    tmpJoin.Cost = (tmpJoin.MessageTrans)*(curBest.TotalSize);
                    tmp = tmpJoin;
                }
                else {
                    tmp = tmpJoin.Children.First();
                }
            }
            else {
                AlgebraTreeNode tmpUnion = new AlgebraTreeNode();
                tmpUnion.Type = AlgebraTreeNodeType.UNION;

                Permutations<Serialization.Fragment> perms = new Permutations<Serialization.Fragment>(table.Fragments);

                int factor = table.Fragments.Count ;
                IEnumerator<IList<Serialization.Fragment>> enumerator = perms.GetEnumerator();

                AlgebraTreeNode curBestPhyPlan = null;
                
                double curBestCost =Double.MaxValue;
                while (enumerator.MoveNext())
                {
                    //logical plan
                    IList<Serialization.Fragment> plan = enumerator.Current;
                    int l = 0, r = factor - 1;
                    List<AlgebraTreeNode> phyPlans= createBestPhysicalPlan(plan, l,r, selection);
                    //catalan(factor+1) physical plan for each logical plan
                    foreach (AlgebraTreeNode phyPlan in phyPlans)
                    {
                        if (phyPlan.Cost < curBestCost)
                        {
                            curBestCost = phyPlan.Cost;
                            curBestPhyPlan = phyPlan;
                        }
                    }
                }


                tmp = curBestPhyPlan;
            
            }

            return tmp;
        }

        // r can be achieved
        private static List<AlgebraTreeNode> createBestPhysicalPlan(IList<Serialization.Fragment> plan, int l, int r, SelectionNode selection)
        {
            
            if (r >l)
            {
                List<AlgebraTreeNode> bestNodes = new List<AlgebraTreeNode>();
                AlgebraTreeNode best = null;
                double bestCost = Double.MaxValue;
                for (int i = l; i < r; ++i)
                {
                    
                    List<AlgebraTreeNode> lNodes = createBestPhysicalPlan(plan, l, i,selection);
                    List<AlgebraTreeNode> rNodes = createBestPhysicalPlan(plan, i+1, r, selection);
                    foreach (AlgebraTreeNode lchild in lNodes)
                    {
                        foreach (AlgebraTreeNode rchild in rNodes)
                        {
                            AlgebraTreeNode node = new AlgebraTreeNode();
                            node.Type = AlgebraTreeNodeType.UNION;
                            node.Children.Add(lchild);
                            node.Children.Add(rchild);
                            lchild.Parent = node;
                            rchild.Parent = node;
                            node.ExecSite =lchild.ExecSite;
                            //rough estimateion
                            
                            node.TotalSize = lchild.TotalSize + rchild.TotalSize;
                            node.MessageTrans = lchild.MessageTrans + rchild.MessageTrans;
                            if (lchild.ExecSite != rchild.ExecSite)
                            {
                                //
                                node.MessageTrans += rchild.TotalSize; 
                            }
                            node.Cost = node.TotalSize + node.MessageTrans;
                            if (node.Cost < bestCost)
                            {
                                bestCost = node.Cost;
                                best = node;
                            }
                        }
                    }
                }
                bestNodes.Add(best);
                return bestNodes;
            }
            else if (r == l)
            {
                List<AlgebraTreeNode> nodes = new List<AlgebraTreeNode>();
                foreach (Serialization.Site site in plan.ElementAt(r).Sites)
                {
                    
                    Serialization.HFragment hFrag =  plan.ElementAt(r) as Serialization.HFragment;
                    FragmentNode fragNode = new FragmentNode(hFrag, AlgebraTreeNodeType.HFRAGMENT);
                    fragNode.ExecSite = site;
                    fragNode.TotalSize = (hFrag.RowCount * hFrag.RowSize);
                    fragNode.MessageTrans = 0;
                    fragNode.Cost = fragNode.TotalSize;
                    if (selection != null)
                    {
                        AlgebraTreeNode selectionNode = selection.dup();
                        selectionNode.ExecSite = fragNode.ExecSite;
                        selection.TotalSize = fragNode.TotalSize;
                        selectionNode.MessageTrans = fragNode.MessageTrans;
                        selectionNode.Cost = fragNode.Cost;
                        fragNode.Parent = selectionNode;
                        selectionNode.Children.Add(fragNode);
                        nodes.Add(selectionNode);
                    }
                    else
                    {
                        nodes.Add(fragNode);
                    }
                }
                return nodes;
                //return new FragmentNode(plan.ElementAt(r), AlgebraTreeNodeType.HFRAGMENT);
            }
            return new List<AlgebraTreeNode>();
            
        }

        private static AlgebraTreeNode createBestPhysicalPlan(IList<AlgebraTreeNode> plan, int l, int r,  List<Serialization.Column> columns)
        {

            if (r > l)
            {
                List<AlgebraTreeNode> bestNodes = new List<AlgebraTreeNode>();
                AlgebraTreeNode best = null;
                double bestCost = Double.MaxValue;
                for (int i = l; i < r; ++i)
                {

                    AlgebraTreeNode lNode = createBestPhysicalPlan(plan, l, i, columns);
                    AlgebraTreeNode rNode = createBestPhysicalPlan(plan, i + 1, r, columns);
                    HashSet<Serialization.Column> lColumns = extractColums(lNode);
                    HashSet<Serialization.Column> rColumns = extractColums(rNode);
                    JoinNode node = new JoinNode();
                    int joinCounts = columns.Count;
                    List<Serialization.Column> hitColumns = new List<Serialization.Column>();
                    for (int j = 0; j < (joinCounts / 2); ++j)
                    {
                        if (lColumns.Contains(columns.ElementAt(j)) && rColumns.Contains(columns.ElementAt(j + 1)))
                        {
                            hitColumns.Add(columns.ElementAt(j));
                            hitColumns.Add(columns.ElementAt(j+1));
                        }else if (lColumns.Contains(columns.ElementAt(j+1)) && rColumns.Contains(columns.ElementAt(j )))
                        {
                            hitColumns.Add(columns.ElementAt(j + 1));
                            hitColumns.Add(columns.ElementAt(j));
                        }
                     }
                     if (hitColumns.Count > 2)
                     {
                         continue;
                     }else if(hitColumns.Count==0)
                     {
                         //product
                         node.Type = AlgebraTreeNodeType.PRODUCT;
                     }
                     else if (hitColumns.Count == 2)
                     {
                                node.Columns.Add(hitColumns.First());
                                node.Columns.Add(hitColumns.ElementAt(1));
                     }

                            node.ExecSite = lNode.ExecSite;
                            node.Children.Add(lNode);
                            node.Children.Add(rNode);
                            lNode.Parent = node;
                            rNode.Parent = node;

                            //rough estimateion
                            node.TotalSize = (lNode.TotalSize) * (rNode.TotalSize);
                            node.MessageTrans = lNode.MessageTrans + rNode.MessageTrans;
                            if (lNode.ExecSite != rNode.ExecSite)
                            {
                                //
                                node.MessageTrans += rNode.TotalSize;
                            }
                            node.Cost = node.TotalSize + node.MessageTrans;
                            if (node.Cost < bestCost)
                            {
                                bestCost = node.Cost;
                                best = node;
                            }
                        
                }
                return best;
            }
            else if (r == l)
            {
                return plan.ElementAt(r);
                //return new FragmentNode(plan.ElementAt(r), AlgebraTreeNodeType.HFRAGMENT);
            }
            return null;

        }

        private static HashSet<Serialization.Column> extractColums(AlgebraTreeNode node)
        {
            HashSet<Serialization.Column> columns = new HashSet<Serialization.Column>();
            switch (node.Type)
            {
                case AlgebraTreeNodeType.HFRAGMENT:
                    FragmentNode hFragNode = node as FragmentNode;
                    Serialization.HFragment hFrag = (Serialization.HFragment)(hFragNode.Frag);
                    Serialization.Table htable = (Serialization.ConfHelper.name2table)[hFrag.TableName];
                    foreach (Serialization.Column c in htable.Columns)
                    {
                        columns.Add(c);
                    }
                    break;
                case AlgebraTreeNodeType.VFRAGMENT:
                    FragmentNode vFragNode = node as FragmentNode;
                    Serialization.VFragment vFrag = (Serialization.VFragment)(vFragNode.Frag);
                    foreach (Serialization.Column c in vFrag.ColumnRefs)
                    {
                        columns.Add(c);
                    }
                    break;
                case AlgebraTreeNodeType.TABLE:
                    TableNode tableNode = node as TableNode;
                    foreach (Serialization.Column c in tableNode.Table.Columns)
                    {
                        columns.Add(c);
                    }

                    break;
                case AlgebraTreeNodeType.PROJECTION:
                    ProjectionNode projectionNode = node as ProjectionNode;
                    foreach (Serialization.Column c in projectionNode.Columns)
                    {
                        columns.Add(c);
                    }
                    break;
                default:
                    foreach (AlgebraTreeNode child in node.Children)
                    {
                        HashSet<Serialization.Column> childColumns = extractColums(child);
                        columns.UnionWith(childColumns);
                    }
                    break;
            }
            return columns;
        }

        private static void collectJoinConds(Serialization.Condition cond, List<Serialization.Column> columns)
        {
            if (Serialization.Condition.isPred(cond))
            {
                if (Serialization.Condition.isJoinCond(cond))
                {
                    HashSet<Serialization.Column> condColumns = Serialization.Condition.getInvolvedColumns(cond);
                    columns.AddRange(condColumns);
                }
               
            }
            
            else if (cond.Type == Serialization.ConditionType.AND )
            {
                foreach (Serialization.Condition child in cond.Children)
                {
                    collectJoinConds(child, columns);
                }
            }
        }

        private static void collectConds(Serialization.Condition cond, Dictionary<string, List<Serialization.Condition>> tableName2Selection)
        {
            if (Serialization.Condition.isPred(cond))
            {
                if (!Serialization.Condition.isJoinCond(cond))
                {
                    HashSet<Serialization.Table> tables = Serialization.Condition.getInvolvedTables(cond);
                    IEnumerator<Serialization.Table> tableEnumerator = tables.GetEnumerator();
                    string selectionTableNmae;
                    while(tableEnumerator.MoveNext())
                    {
                        selectionTableNmae = tableEnumerator.Current.Name;
                        if (!tableName2Selection.ContainsKey(selectionTableNmae))
                        {
                            tableName2Selection[selectionTableNmae] = new List<Serialization.Condition>();
                        }
                        tableName2Selection[selectionTableNmae].Add(cond);
                    }
                    
                }
                else
                {
                    //join , columns may be used in projection 

                }
            }
            else if (cond.Type == Serialization.ConditionType.OR)
            {
                HashSet<Serialization.Table> tables = Serialization.Condition.getInvolvedTables(cond);
                if(tables.Count == 1)
                {
                    IEnumerator<Serialization.Table> tableEnumerator = tables.GetEnumerator();
                    string selectionTableNmae;
                    while(tableEnumerator.MoveNext())
                    {
                        selectionTableNmae = tableEnumerator.Current.Name;
                        if (!tableName2Selection.ContainsKey(selectionTableNmae))
                        {
                            tableName2Selection[selectionTableNmae] = new List<Serialization.Condition>();
                        }
                        tableName2Selection[selectionTableNmae].Add(cond);
                    }
                }
            }
            else if (cond.Type == Serialization.ConditionType.AND)
            {
                foreach (Serialization.Condition child in cond.Children)
                {
                    collectConds(child, tableName2Selection);
                }
            }
        }

        private static void determineExecSite(AlgebraTreeNode node)
        {
            if (node.ExecSite != null)
            {
                return;
            }
            else {
                determineExecSite(node.Children.First());
                node.ExecSite = node.Children.First().ExecSite;
            }
        }

        private static QEMessage buildMessagesHelper(AlgebraTreeNode node, List<QEMessage> messages)
        {
            QEMessage message = new QEMessage();
            message.CommandType = 0;
            message.DestinationName = node.ExecSite.SiteName;
            if (node.Parent == null)
            {
                //final
                message.NextDestinationName = null;
            }else{
                message.NextDestinationName = node.Parent.ExecSite.SiteName;
            }
            message.QueryID = new QueryID();
            message.QueryID.CurrentID = QueryID.GetNewGuid();
            QEMessage subMsg=null;
            List<QEMessage> subMsgs=null;
            StringBuilder builder =null;
            IList<Guid> childIds = null;
            bool newMsg = true;
            switch (node.Type)
            {
                    
                case AlgebraTreeNodeType.HFRAGMENT:
                    //sql
                    FragmentNode hFragNode = node as FragmentNode;
                    message.OperationType = 0;
                    builder = new StringBuilder();
                    builder.Append("select * from ");
                    builder.Append(hFragNode.Frag.TableName);
                    if (hFragNode.Parent != null && hFragNode.Parent.Type == AlgebraTreeNodeType.SELECTION)
                    {
                        builder.Append(" where ");
                        builder.Append(((hFragNode.Parent as SelectionNode).Cond).toString());
                        builder.Append(";");
                    }
                    message.CommandString = builder.ToString();
                    message.TableName = hFragNode.Frag.TableName;
                    break;
                case AlgebraTreeNodeType.VFRAGMENT:
                    //sql
                    FragmentNode vFragNode = node as FragmentNode;
                    message.OperationType = 0;
                    builder = new StringBuilder();
                    builder.Append("select * from ");
                    builder.Append(vFragNode.Frag.TableName);
                    if (vFragNode.Parent != null && vFragNode.Parent.Type == AlgebraTreeNodeType.SELECTION)
                    {
                        builder.Append(" where ");
                        builder.Append(((vFragNode.Parent as SelectionNode).Cond).toString());
                        builder.Append(";");
                    }
                    message.CommandString = builder.ToString();
                    message.OperationType = 0;
                    message.TableName = vFragNode.Frag.TableName;
                    break;
                case AlgebraTreeNodeType.SELECTION:
                    message.OperationType = 4;
                    subMsg = buildMessagesHelper(node.Children.First(), messages);
                    if (node.Children.First().Type != AlgebraTreeNodeType.HFRAGMENT)
                    {
                        message.QueryID.ChildID = new Guid[1];
                        message.QueryID.ChildID[0] = subMsg.QueryID.CurrentID;
                        message.CommandString = (node as SelectionNode).Cond.ToString();

                    }
                    else {
                        newMsg = false;
                        message = subMsg;
                    }

                    break;
                case AlgebraTreeNodeType.PROJECTION:
                    message.OperationType = 3;
                    subMsg = buildMessagesHelper(node.Children.First(), messages);
                    message.QueryID.ChildID = new Guid[1];
                    message.QueryID.ChildID[0] = subMsg.QueryID.CurrentID;
                    IList<Serialization.Column> projectionColumns = (node as ProjectionNode).Columns;
                    IList<string> projectionColumnNames = new List<string>();
                    foreach(Serialization.Column projectionColumn in projectionColumns)
                    {
                        projectionColumnNames.Add(projectionColumn.TableName+"_"  +projectionColumn.ColumnName);
                    }
                    message.columnNames = projectionColumnNames.ToArray();

                    break;
                case AlgebraTreeNodeType.UNION:
                    subMsgs = new List<QEMessage>();
                    foreach(AlgebraTreeNode child in node.Children)
                    {
                        subMsgs.Add(buildMessagesHelper(child, messages));
                    }
                    childIds = new List<Guid>();
                    foreach(QEMessage child in subMsgs)
                    {
                        childIds.Add(child.QueryID.CurrentID);
                    }
                    message.QueryID.ChildID = childIds.ToArray();
                    message.OperationType = 5;
                    break;
                case AlgebraTreeNodeType.JOIN:
                    message.OperationType = 1;
                    subMsgs = new List<QEMessage>();
                    foreach(AlgebraTreeNode child in node.Children)
                    {
                        subMsgs.Add(buildMessagesHelper(child, messages));
                    }
                    childIds = new List<Guid>();
                    foreach(QEMessage child in subMsgs)
                    {
                        childIds.Add(child.QueryID.CurrentID);
                    }
                    message.QueryID.ChildID = childIds.ToArray();
                    JoinNode joinNode = node as JoinNode;
                    IList<string> joinColumnNames = new List<string>();
                    foreach(Serialization.Column joinColumn in joinNode.Columns)
                    {
                        joinColumnNames.Add(joinColumn.ColumnName);
                    }
                    message.columnNames = joinColumnNames.ToArray();
                    break;
                case AlgebraTreeNodeType.PRODUCT:
                    
                    message.OperationType = 2;
                    subMsgs = new List<QEMessage>();
                    foreach(AlgebraTreeNode child in node.Children)
                    {
                        subMsgs.Add(buildMessagesHelper(child, messages));
                    }
                    childIds = new List<Guid>();
                    foreach(QEMessage child in subMsgs)
                    {
                        childIds.Add(child.QueryID.CurrentID);
                    }
                    message.QueryID.ChildID = childIds.ToArray();
                    break;
                default: 
                    break;
            }
            if (newMsg)
            {
                messages.Add(message);
            }
            return message;
        }

        public List<QEMessage> buildMessages(string sitename)
        {
            List<QEMessage> messages = new List<QEMessage>();
            buildMessagesHelper(root, messages);
            
            if (messages.Last().NextDestinationName == null)
            {
                if (messages.Last().DestinationName != sitename)
                {
                    //messages.Last().DestinationName = sitename;
                    QEMessage last = new QEMessage();
                    last.QueryID = new QueryID();
                    last.QueryID.CurrentID = QueryID.GetNewGuid();
                    last.QueryID.ChildID = new Guid[1];
                    last.QueryID.ChildID[0] = messages.Last().QueryID.CurrentID;
                    last.DestinationName = sitename;
                    messages.Last().NextDestinationName = sitename;
                    last.CommandType = 0;
                    last.OperationType = -1; //nop
                    
                    

                }
            }
            Guid querySessionId = QueryID.GetNewGuid();
            foreach (QEMessage msg in messages)
            {
                msg.QueryID.ParentQueryID = querySessionId;
            }
            return messages;
        }

        private void toStringHelper(AlgebraTreeNode node, StringBuilder builder,int nspace)
        {
            for(int i=0; i<nspace;++i)
            {
                builder.Append(" ");
            }
            builder.Append(node.ExecSite.SiteName+" : ");
            switch (node.Type)
            {
                case AlgebraTreeNodeType.JOIN:
                    JoinNode joinNode = node as JoinNode;
                    
                    builder.Append("join: ");
                    builder.Append(joinNode.Columns[0].ColumnName);
                    builder.Append(" = ");
                    builder.Append(joinNode.Columns[1].ColumnName);
                    builder.Append("\r\n");
                    foreach(AlgebraTreeNode child in node.Children)
                    {
                        toStringHelper(child, builder, nspace+4);
                    }
                    break;
                case AlgebraTreeNodeType.UNION:
                    builder.Append("union:");
                    builder.Append("\r\n");
                    foreach(AlgebraTreeNode child in node.Children)
                    {
                        toStringHelper(child, builder, nspace+4);
                    }
                    break;
                case AlgebraTreeNodeType.SELECTION:
                    SelectionNode selectionNode = node as SelectionNode;
                    
                    builder.Append("selection: ");
                    builder.Append(selectionNode.Cond.ToString());
                    builder.Append("\r\n");
                    toStringHelper(selectionNode.Children[0], builder, nspace+4);
                    
                    break;
                case AlgebraTreeNodeType.PRODUCT:
                    builder.Append("cartesian product:");
                    builder.Append("\r\n");
                    foreach(AlgebraTreeNode child in node.Children)
                    {
                        toStringHelper(child, builder, nspace+4);
                    }
                    break;
                case AlgebraTreeNodeType.PROJECTION:
                    ProjectionNode projectionNode = node as ProjectionNode;
                    
                    builder.Append("projection: ");
                    foreach(Serialization.Column c in projectionNode.Columns)
                    {
                        builder.Append(c.ColumnName);
                        builder.Append(" ");
                    }
                    builder.Append("\r\n");
                    toStringHelper(projectionNode.Children[0], builder, nspace + 4);
                    break;
                case AlgebraTreeNodeType.VFRAGMENT:
                    builder.Append((node as FragmentNode).Frag.TableName);
                    builder.Append("\r\n");
                    break;
                case AlgebraTreeNodeType.HFRAGMENT:
                    builder.Append((node as FragmentNode).Frag.TableName);
                    builder.Append("\r\n");
                    break;
                default:
                    break;
            }
        }

        override public string ToString()
        {
            StringBuilder builder = new StringBuilder();
            toStringHelper(root, builder,0);
            return builder.ToString();
        }

    }
}
